import com.bbq.framework.Loader;
import com.bbq.job.JobProcessor;
import com.bbq.job.JobResponse;
import com.bbq.job.JobSpec;
import com.bbq.job.WorkerConfig;
import com.bbq.model.common.JobModel;
import com.bbq.model.common.JobUpdateModel;
import com.bbq.service.job.JobService;
import com.bbq.utils.ConfigUtil;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.AbstractModule;
import com.google.inject.Injector;
import com.google.inject.Singleton;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.MessageProperties;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.concurrent.Executors;
import org.apache.log4j.Logger;
import org.bouncycastle.util.Strings;
import worker.config.WorkerConfigImpl;

public class Main {
	static Logger logger = Logger.getLogger(Main.class);

	public static void main(String[] args) throws Exception {
		String hostName = ConfigUtil.getString("rabbitmq.host", "192.168.199.222");
		Integer portNumber = ConfigUtil.getInteger("rabbitmq.port", null);
		String userName = ConfigUtil.getString("rabbitmq.user", null);
		String password = ConfigUtil.getString("rabbitmq.password", null);
		String virtualHost = ConfigUtil.getString("rabbitmq.virtual_host", null);
		ObjectMapper mapper = new ObjectMapper();

		ConnectionFactory factory = new ConnectionFactory();
		factory.setAutomaticRecoveryEnabled(true);

		factory.setHost(hostName);
		if (portNumber != null) {
			factory.setPort(portNumber);
		}
		if (userName != null) {
			factory.setUsername(userName);
		}
		if (password != null) {
			factory.setPassword(password);
		}
		if (virtualHost != null) {
			factory.setVirtualHost(virtualHost);
		}

		Map<String, JobProcessor> jobProcessors = Maps.uniqueIndex(getJobProcessors(), jp -> jp.getJobClass());
		ListeningExecutorService executors = MoreExecutors
				.listeningDecorator(Executors.newFixedThreadPool(ConfigUtil.getInteger("worker.threads", 10)));

		// --- setup ---
		final Connection conn = factory.newConnection();
		final Channel channel = conn.createChannel();

		String queueName = ConfigUtil.getString("rabbitmq.jobqueue", "bf.JobQueue");
		String updateName = ConfigUtil.getString("rabbitmq.jobupdate", "bf.JobUpdate");

		channel.queueDeclare(queueName, true, false, false, null);
		logger.info(">>>> Waiting for Job");

		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {

				try {
					logger.info("request >> " + Strings.fromUTF8ByteArray(body));
					JsonNode n = mapper.readTree(Strings.fromUTF8ByteArray(body));
					JobModel jobModel = mapper.convertValue(n, JobModel.class);
					logger.info("Received Job: " + jobModel.getId() + " (" + jobModel.getJobClass() + "): "
							+ jobModel.getJobParams());

					final JobProcessor jp = jobProcessors.get(jobModel.getJobClass());
					if (jp != null) {
						ListenableFuture<JobResponse> result = executors.submit(() -> {
							final int jobID = jobModel.getId();
							JobSpec spec = new JobSpec(jobModel.getJobClass(), jobModel.getJobParams());
							try {
								return jp.process(spec, progress -> {
									try {
										JobUpdateModel upd = new JobUpdateModel();
										upd.setJobId(jobModel.getId());
										upd.setProgress(progress);
										updateProgress(factory, updateName, mapper, upd);
									} catch (Exception e) {
										logger.error("Error Updating Progress", e);
									}
								});
							} finally {
								logger.info("Finished Job: " + jobID);
							}
						});

						Futures.addCallback(result, new FutureCallback<JobResponse>() {
							@Override
							public void onFailure(Throwable t) {
								logger.error("Job Response Error Job: " + jobModel.getId() + " ("
										+ jobModel.getJobClass() + "): " + jobModel.getJobParams(), t);
								JobUpdateModel upd = new JobUpdateModel();
								upd.setJobId(jobModel.getId());
								upd.setError(t.getMessage());
								upd.setProgress(100);
								upd.setDone(true);
								upd.setResult(null);
								upd.setResultType(null);
								updateProgress(factory, updateName, mapper, upd);
								try {
									if (!(t instanceof RuntimeException)) {
										channel.basicAck(envelope.getDeliveryTag(), false);
									} else {
										logger.error("Runtime Error, Message Not Ack");
									}
								} catch (IOException e) {
									logger.error("Ack Error Job: " + jobModel.getId() + " (" + jobModel.getJobClass()
											+ "): " + jobModel.getJobParams(), e);
								}
							}

							@Override
							public void onSuccess(JobResponse resp) {
								JobUpdateModel upd = new JobUpdateModel();
								upd.setJobId(jobModel.getId());
								upd.setProgress(100);
								upd.setDone(true);
								upd.setResult(resp.getContent());
								upd.setResultType(resp.getContentType());
								updateProgress(factory, updateName, mapper, upd);
								try {
									channel.basicAck(envelope.getDeliveryTag(), false);
								} catch (IOException e) {
									logger.error("Ack Error Job: " + jobModel.getId() + " (" + jobModel.getJobClass()
											+ "): " + jobModel.getJobParams(), e);
								}
							}
						});

					} else {
						logger.error("Job not handled: " + jobModel.getJobClass());
					}
				} catch (Exception e) {
					logger.error("Error Processing Job", e);
				}

			}
		};
		channel.basicConsume(queueName, false, consumer);
	}

	protected static void updateProgress(ConnectionFactory factory, String topicName, ObjectMapper mapper,
			JobUpdateModel update) {
		try {
			final Connection conn = factory.newConnection();
			final Channel channel = conn.createChannel();
			channel.exchangeDeclare(topicName, "fanout");
			byte[] body = mapper.writeValueAsBytes(update);
			channel.basicPublish(topicName, "", MessageProperties.PERSISTENT_TEXT_PLAIN, body);
			channel.close();
			conn.close();
		} catch (Exception e) {
			logger.error("Error Updating Progress", e);
		}
	}

	protected static List<JobProcessor> getJobProcessors() {
		ServiceLoader<JobProcessor> processors = ServiceLoader.load(JobProcessor.class);
		List<JobProcessor> jps = Lists.newArrayList(processors.iterator());
		
		Injector injector = Loader.getInstance().loadIfNotLoaded(new AbstractModule() {
			@Override
			protected void configure() {
				bind(JobService.class).toInstance(JobService.getInstance("job-worker"));
				bind(WorkerConfig.class).to(WorkerConfigImpl.class).in(Singleton.class);
			}
		});
		for (JobProcessor jp : jps) {
			logger.info("JobProcessor>>>:" + jp.getJobClass());
			injector.injectMembers(jp);
		}
		return jps;
	}

	public static String getLocalHostName() {
		try {
			return InetAddress.getLocalHost().getHostName();
		} catch (UnknownHostException e) {
			logger.error("ERROR", e);
			return "UNKNOWN";
		}
	}
}
